/*
	Copyright (c) 2008, Grant Jones <grant@podtronics.com>.
	All rights reserved.

	Redistribution and use in source and binary forms, with or
	without modification, are permitted provided that the following
	conditions are met:

		Redistributions of source code must retain the above copyright
		notice, this list of conditions and the following disclaimer.
		
		Redistributions in binary form must reproduce the above copyright
		notice, this list of conditions and the following disclaimer in
		the documentation and/or other materials provided with the
		distribution.
		
		Neither the name of the gxjones.com nor the names of its
		contributors may be used to endorse or promote products derived
		from this software without specific prior written permission.
	
	THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
	"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
	LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
	FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
	COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
	INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
	(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
	SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
	HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
	STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
	ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
	OF THE POSSIBILITY OF SUCH DAMAGE.
	
	http://www.opensource.org/licenses/bsd-license.php
*/

/*
 *  tagmt.c
 *  radixtree
 *
 *  Created by Grant Jones on 2/29/08.
 *
 */
 
/*
** Tag Transaction Management Routines
*/

/* TODO: 
** - a read or write and then a subsequent read or write partially over a
** cached region will most likely cause issues
**
** - NEEDs to implement a LRU type cache
*/

#include "tagm.h"
#include "tagmt.h"
#include "cmap.h"	/* cmap will work for db sizes < 0x80000000 (as of 2/29/08) */


void tagmt_init( TagManagerStruct *tags )
{
	tags->transactions = NULL;
}

/* ************************************************************************
************************************************************************ */

/*!
	@function tagmt_popTransaction
	@abstract Internal convenience function
	@discussion 
	@result pops the most recent transaction and returns it
*/
TagManagerTrans *tagmt_popTransaction( TagManagerStruct *tags )
{
	TagManagerTrans *t = (TagManagerTrans *)tags->transactions;
	if(!t) return NULL;
	tags->transactions = t->link;
	
	return t;
}

/*!
	@function tagmt_cached_read_get
	@abstract Internal convenience function
	@discussion Used ONLY for tagmt_pread's where we might be an embedded
		transaction situation, searches all active transactions.
	@result returns the cache item, NULL if error or not found.
*/
TagManagerTC *tagmt_cached_read_get( TagManagerStruct *tags, size_t nbyte, off_t offset )
{
	TagManagerTrans *t;
	for( t = (TagManagerTrans *)tags->transactions; t; t = t->link )
	{
		TagManagerTC *c = cmap_lookup(t->cache, (CMAP_KEY_TYPE)offset );
		if(!c)
			continue;
		assert( c->offset == offset );
		
		assert( c->nbyte >= nbyte );
		return c;
	}
	return NULL;
}

/*!
	@function tagmt_cached_write_get
	@abstract Internal convenience function
	@discussion Used ONLY for tagmt_pwrites, only searches the top transaction.
	@result returns the cache item, NULL if error or not found.
*/
TagManagerTC *tagmt_cached_write_get(TagManagerTrans * t, size_t nbyte, off_t offset )
{
	if(!t) return NULL;

	TagManagerTC *c = cmap_lookup( t->cache, (CMAP_KEY_TYPE)offset );
	if(!c)
		return NULL;

	assert( c->offset == offset );
	assert( c->nbyte >= nbyte );	/* TODO: resize the cached item in this case */

	return c;
}

/*!
	@function tagmt_cached_new
	@abstract Internal convenience function
	@discussion Creates a new cache item in the specified transaction.
	@param t
	@param buf
	@param nbyte
	@param offset
	@result returns the new cache item, NULL on error.
*/
TagManagerTC *tagmt_cached_new( TagManagerTrans * t, void *buf, size_t nbyte, off_t offset )
{
	if(!t) return NULL;

	TagManagerTC *tc = (TagManagerTC *)cmap_assign_new(t->cache, (CMAP_KEY_TYPE)offset, sizeof(TagManagerTC) + nbyte);
	if(!tc)
		return NULL;

	tc->writeCount = 0;
	tc->offset = offset;
	tc->nbyte = nbyte;
	memcpy( tc->data, buf, nbyte );
	
	return tc;
}

/* ************************************************************************
************************************************************************ */


/*!
	@function tagmt_beginTransaction
	@abstract currently a no-op
	@discussion in the future this will begin a group of write operations
	@result returns the current transaction ID; -1 on error
*/
int tagmt_beginTransaction( TagManagerStruct *tags )
{
	TagManagerTrans *t = (TagManagerTrans *)malloc( sizeof(TagManagerTrans) );
	if(!t)
		return -1;
	memset( t, 0, sizeof(TagManagerTrans) );
	
	t->link = tags->transactions;
	t->tid = (t->link ? (t->link->tid)+1 : 1 );

//	t->cache = cmap_new( t->link? ((cmapStruct *)t->link->cache)->mempool : NULL );
	t->cache = cmap_new( NULL );
	
	tags->transactions = t;
#ifdef TAGMT_TRACE_STDOUT	
	printf("beginTransaction #%d\n", t->tid);
#endif
	
	return t->tid;
}

/*!
	@function tagmt_endTransaction
	@abstract currently a no-op
	@discussion Commit a group of write operations. If there are multiple
		transactions this will cascade the writes to the next active
		transaction. If this is transaction #1 then it will do an actual
		write (producing a rollback in this case TODO still).
	@param tags current tag context
	@param tid Transaction ID provided by the begin transaction operation	
	@result returns 0 on success; -1 on error
*/
int tagmt_endTransaction( TagManagerStruct *tags, int tid )
{
	TagManagerTC *c;
//	TagManagerTC *c_next;
	TagManagerTrans *t = tagmt_popTransaction( tags );
	int failure = 0;
	if(!t)
		return -1;
	if( t->tid != tid )
	{
		TAGM_ERROR(("%s Out of Order: Cannot end transaction #%d before ending #%d\n", __FUNCTION__, tid, t->tid ));
		/* This is the result of improper tag transaction use */
		assert(0);		
		return -1;
	}
#ifdef TAGMT_TRACE_STDOUT	
	printf("endTransaction #%d\n", tid);
#endif	

//	if( tid == 1 )
//		printf("endTransaction #%d; cache items: #%ld\n", tid, ((cmapStruct *)t->cache)->num );


	cmapEnumStruct e;
	CMAP_KEY_TYPE key;
	int journal_fd = -1;
	char journal_template[] = "journal.XXXXXX";
	cmap_enum( t->cache, &e );

/* Enable this to do journaling:
	#define TAGMT_WRITE_JOURNAL
*/
#ifdef TAGMT_WRITE_JOURNAL	
	if( !tags->transactions && t->writeCount > 0 )
	{
		journal_fd = tagmt_journal_begin(journal_template);
		if(journal_fd < 0)
		{
			fprintf(stderr, "%s Failed to create journal file before transaction commit, errno: %d\n", __FUNCTION__, errno );
			return -1;
		}
	}
#endif
	
	while( (c = cmap_enum_next( &e, &key )) )
	{
#ifdef TAGMT_TRACE_STDOUT		
		printf("nbyte: %ld; offset: %ld\n", (int)c->nbyte, (int)c->offset );
#endif		
		/* write out only if we need to */
		if( c->writeCount > 0 )
		{
			if( journal_fd > 0 )
			{
				ssize_t read_sz;
				void *org_data = mempool_malloc_named( ((cmapStruct *)t->cache)->mempool, c->nbyte, "journal original data" );
				/* read the data we will be over writing: */
				read_sz = tagmt_pread(tags, org_data, c->nbyte, c->offset );
				if( read_sz != 0 )
				{
					if( read_sz != c->nbyte )
					{
						failure = -1;
						break;
					}
					/* write to the journal: */
					if( tagmt_journal_write( journal_fd, org_data, c->data, c->nbyte, c->offset ) )
					{
						fprintf(stderr, "%s Journal file write failed: %d\n", __FUNCTION__, errno );
						failure = -1;
						break;
					}
				}
				/*
				else
				{
					Ignore reads of zero, this was probably an append to the end of the file
				}
				*/
			}
			
			if( tagmt_pwrite(tags, c->data, c->nbyte, c->offset) != c->nbyte )
			{
				failure = -1;
				break;
			}
/*#define RANDOM_JOURNAL_TEST*/
#ifdef RANDOM_JOURNAL_TEST
			if( random()%2 == 1 )
			{
				failure = -1;
				break;
			}
#endif
		}
	}		

	cmap_destroy( t->cache );
	
	free(t);

	if( failure )
		tagmt_journal_rollback(journal_fd, tags);
	
	tagmt_journal_end(journal_template, journal_fd);	
	return failure;
}

/*!
	@function tagmt_abortTransaction
	@abstract currently a no-op
	@discussion in the future this will abort a group of write operations
	@param tags current tag context
	@param tid Transaction ID provided by the begin transaction operation
	@result returns 0 on success; -1 on error
*/
int tagmt_abortTransaction( TagManagerStruct *tags, int tid )
{
	assert( tid > 0 );
	TagManagerTrans *t = (TagManagerTrans *)tags->transactions;
	if(!t) return 0;
	if( tid != t->tid )	/* the transaction may already have been popped by tagmt_endTransaction, in this case just return */
		return 0;
		
	tags->transactions = t->link;
			
	if( t->tid != tid )
	{
		TAGM_ERROR(("%s Out of Order: Cannot end transaction #%d before ending #%d\n", __FUNCTION__, tid, t->tid ));
		/* This is the result of improper tag transaction use */
		assert(0);
		return -1;
	}	
	
//	TAGM_TRACE(("%s",__FUNCTION__));

	cmap_destroy( t->cache );
	free(t);
	
	return 0;
}
/* ************************************************************************
************************************************************************ */
size_t tagmt_pread( TagManagerStruct *tags, void *buf, size_t nbyte, off_t offset )
{
	TagManagerTC *tc;
	size_t r;
#ifdef TAGM_DEBUG_NAME
	printf("tagmt_pread nbyte: %ld offset: %ld fd: %d filename: %s\n", (int)nbyte, (int)offset, tags->fd, tags->name );
#endif
	if( nbyte == 0 ) return 0;
	

	if( nbyte > MAX_CACHE_ITEM || !tags->transactions )
	{
		if( tags->fd >= 0 )
		{
			r = pread( tags->fd, buf, nbyte, offset);
		}
		else
		{
			r = nbyte;
			memset( buf, 0, nbyte );
		}
			
		if( r != nbyte )
		{
			/*printf("tagmt_pread nbyte: %ld offset: %ld pread returned: %ld\n", (long int)nbyte, (long int)offset, (long int)r );*/
			
			
		}
		return r;
	}

	tc = tagmt_cached_read_get( tags, nbyte, offset );
	if( !tc )
	{
		/* cache miss */
		if( tags->fd >= 0 )
		{
			r = pread( tags->fd, buf, nbyte, offset);
		}
		else
		{
			memset( buf, 0, nbyte );
			r = nbyte;
		}
		
		if( r != nbyte )	/* will not cache an errored read */
		{
//			printf("read size differs from requested size\n");
//			printf("tagmt_pread nbyte: %ld offset: %ld\n", (int)nbyte, (int)offset );
			return r;
		}
		
		tc = tagmt_cached_new( (TagManagerTrans *)tags->transactions, buf, nbyte, offset );
		return nbyte;
	}
	else
	{
		memcpy( buf, tc->data, nbyte );
		return nbyte;
	}
}

size_t tagmt_pwrite( TagManagerStruct *tags, void *buf, size_t nbyte, off_t offset )
{
	TagManagerTC *tc;
	TagManagerTrans *tm = (TagManagerTrans *)tags->transactions;
	
//	printf("tagmt_pwrite nbyte: %ld offset: %ld\n", (int)nbyte, (int)offset );
	
	if( nbyte > MAX_CACHE_ITEM || !tm )
		return pwrite( tags->fd, buf, nbyte, offset);
	
	tc = tagmt_cached_write_get( tm, nbyte, offset );
	if( !tc )
	{
		/* cache miss */
		tc = tagmt_cached_new( tm, buf, nbyte, offset );
		tc->writeCount++;
		tm->writeCount++;
		return nbyte;		
	}
	else
	{
		memcpy( tc->data, buf, nbyte );
		tc->writeCount++;
		tm->writeCount++;
		return nbyte;
	}	
}

/* ************************************************************************
************************************************************************ */
int tagmt_journal_begin( char *journal_template )
{
	/* open journal file, return fd */
	return mkstemp(journal_template);
}

int tagmt_journal_end( char *journal_template, int journal_fd )
{
	if(journal_fd < 0)
		return 0;
	close(journal_fd);
	unlink( journal_template );
	return 0;
}

off_t tagmt_journal_checksum( unsigned char *buf, size_t nbyte )
{
	size_t i;
	off_t checksum = 0;
	
	for( i = 0; i < nbyte; i ++ )
		checksum += buf[i];	
	return checksum;
}

int tagmt_journal_write( int journal_fd, void *old_buf, void *new_buf, size_t nbyte, off_t offset )
{
	ssize_t r;
	off_t checksum = tagmt_journal_checksum( old_buf, nbyte );
	
	r = write(journal_fd, &nbyte, sizeof(size_t) ) - sizeof(size_t);
	r += write(journal_fd, &offset, sizeof(off_t) ) - sizeof(off_t);
	r += write(journal_fd, old_buf, nbyte ) - nbyte;
	r += write(journal_fd, new_buf, nbyte ) - nbyte;
	r += write(journal_fd, &checksum, sizeof(checksum) ) - sizeof(checksum);
	
	return r;
}


int tagmt_journal_rollback( int journal_fd, TagManagerStruct *tags )
{
	size_t nbyte;
	off_t offset;
	off_t checksum;
	void *d = NULL;
	int result = -1;
	
	if(journal_fd < 0)	/* can't rollback without a journal file */
		return -1;
	
	if( lseek(journal_fd, 0ULL, SEEK_SET ) != 0ULL)
	{
		fprintf(stderr, "Journal rollback failed : seek to journal begining failed, %s\n", strerror(errno) );
		return -1;
	}
	fprintf(stderr, "** Journal rollback starts **\n");
	
	while( read(journal_fd, &nbyte, sizeof(size_t) ) == sizeof(size_t) )
	{
		if( read(journal_fd, &offset, sizeof(off_t) ) != sizeof(off_t) )
		{
			fprintf(stderr, "Journal rollback failed : read offset failed, %s\n", strerror(errno) );
			goto cleanup;
		}
				
		d = malloc( nbyte );
		
		if( read( journal_fd, d, nbyte ) != nbyte )
		{
			fprintf(stderr, "Journal rollback failed : read data failed, %s, nbyte: %ld, offset: %ld\n", strerror(errno), (long)nbyte, (long)offset );
			goto cleanup;
		}
		
		fprintf(stderr, "Journal: rolling back nbyte: %ld, offset: %ld\n", (long)nbyte, (long)offset );


		/* lseek over new data: */
		if( lseek(journal_fd, nbyte, SEEK_CUR) == -1 )
		{
			fprintf(stderr, "Journal rollback failed : seek over new data failed, %s, nbyte: %ld, offset: %ld\n", strerror(errno), (long)nbyte, (long)offset );
			goto cleanup;
		}
		
		if( read( journal_fd, &checksum, sizeof(checksum) ) != sizeof(checksum) )
		{
			fprintf(stderr, "Journal rollback failed : couldn't read checksum, %s\n", strerror(errno) );
			goto cleanup;
		}

		/* checksum old data: */
		if( tagmt_journal_checksum(d, nbyte) != checksum )
		{
			fprintf(stderr, "Journal rollback failed : journal file integrity test failed (got: %x, expected: %x)\n", (unsigned int)checksum, (unsigned int)tagmt_journal_checksum(d, nbyte) );
			goto cleanup;
		}
		

		/* do actual write to the tags file: */
		if( tagmt_pwrite(tags, d, nbyte, offset) != nbyte )
		{
			fprintf(stderr, "Journal rollback failed : write to original file error, %s, nbyte: %ld, offset: %ld\n", strerror(errno), (long)nbyte, (long)offset );
			goto cleanup;
		}
		
									
		
		free(d);
		d = NULL;
		
	}
	
	result = 0;
	
cleanup:
	fprintf(stderr, "** Journal rollback ends %s **\n", (result == 0?"successfully":"unsuccessfully") );
	if(d)
		free(d);
	
	return result;
}








